package com.itheima.behavior.service.impl;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.itheima.apis.book.IBookClient;
import com.itheima.behavior.config.KafkaConfig;
import com.itheima.behavior.service.RankingService;
import com.itheima.common.redis.CacheService;
import com.itheima.model.behavior.dtos.RankingQueryDto;
import com.itheima.model.behavior.vos.RankingItemVo;
import com.itheima.model.common.dtos.ResponseResult;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.stereotype.Service;

import java.util.*;
import java.util.concurrent.TimeUnit;
import java.util.stream.Collectors;

/**
 * 排行榜服务实现类
 * 从 Redis ZSet 读取 Kafka Streams 计算的排行榜数据
 */
@Service
@Slf4j
public class RankingServiceImpl implements RankingService {

    @Autowired
    private CacheService cacheService;
    
    @Autowired
    private IBookClient bookClient;

    @Value("${ranking.default-size:100}")
    private Integer defaultSize;

    @Value("${ranking.hot-size:50}")
    private Integer hotSize;

    // ========== 实时排行榜 ==========

    @Override
    public ResponseResult getRealtimeHotRanking(Integer limit) {
        return getRankingFromRedis(KafkaConfig.Keys.BOOK_REALTIME_HOT_RANKING, limit, "book");
    }

    // ========== 综合排行榜 ==========

    @Override
    public ResponseResult getBookHotRanking(Integer limit, String timeRange) {
        return getRankingFromRedis(KafkaConfig.Keys.BOOK_HOT_RANKING, limit, "book");
    }

    @Override
    public ResponseResult getBookReadRanking(Integer limit, String timeRange) {
        return getRankingFromRedis(KafkaConfig.Keys.BOOK_READ_RANKING, limit, "book");
    }

    @Override
    public ResponseResult getBookCollectRanking(Integer limit, String timeRange) {
        return getRankingFromRedis(KafkaConfig.Keys.BOOK_COLLECT_RANKING, limit, "book");
    }

    @Override
    public ResponseResult getBookLikeRanking(Integer limit, String timeRange) {
        return getRankingFromRedis(KafkaConfig.Keys.BOOK_LIKE_RANKING, limit, "book");
    }

    @Override
    public ResponseResult getUserReadingTimeRanking(Integer limit, String timeRange) {
        return getRankingFromRedis(KafkaConfig.Keys.USER_READING_TIME_RANKING, limit, "user");
    }

    @Override
    public ResponseResult getUserActiveRanking(Integer limit, String timeRange) {
        return getRankingFromRedis(KafkaConfig.Keys.USER_ACTIVE_RANKING, limit, "user");
    }

    @Override
    public ResponseResult getRanking(RankingQueryDto dto) {
        try {
            String rankingType = dto.getRankingType();
            Integer limit = dto.getLimit() != null ? dto.getLimit() : defaultSize;

            // 根据排行榜类型获取对应的 Redis Key
            String redisKey = getRankingKey(rankingType);
            if (redisKey == null) {
                return ResponseResult.errorResult(400, "不支持的排行榜类型");
            }

            String itemType = rankingType.startsWith("book") ? "book" : "user";
            return getRankingFromRedis(redisKey, limit, itemType);

        } catch (Exception e) {
            log.error("获取排行榜失败", e);
            return ResponseResult.errorResult(500, "获取排行榜失败");
        }
    }

    @Override
    public ResponseResult getUserRank(Integer userId, String rankingType) {
        try {
            String redisKey = getRankingKey(rankingType);
            if (redisKey == null) {
                return ResponseResult.errorResult(400, "不支持的排行榜类型");
            }

            // 获取用户排名（从0开始，需要+1）
            Long rank = cacheService.zRank(redisKey, userId);
            if (rank == null) {
                return ResponseResult.okResult(buildRankInfo(null, null, userId, "user"));
            }

            // 获取用户分数
            Double score = cacheService.zScore(redisKey, userId);

            Map<String, Object> rankInfo = buildRankInfo(rank + 1, score, userId, "user");
            return ResponseResult.okResult(rankInfo);

        } catch (Exception e) {
            log.error("获取用户排名失败: userId={}, rankingType={}", userId, rankingType, e);
            return ResponseResult.errorResult(500, "获取用户排名失败");
        }
    }

    @Override
    public ResponseResult getBookRank(Integer bookId, String rankingType) {
        try {
            String redisKey = getRankingKey(rankingType);
            if (redisKey == null) {
                return ResponseResult.errorResult(400, "不支持的排行榜类型");
            }

            // 获取图书排名（倒序排名，需要用 zReverseRank）
            Long rank = cacheService.zReverseRank(redisKey, String.valueOf(bookId));
            if (rank == null) {
                return ResponseResult.okResult(buildRankInfo(null, null, bookId, "book"));
            }

            // 获取图书分数
            Double score = cacheService.zScore(redisKey, String.valueOf(bookId));

            Map<String, Object> rankInfo = buildRankInfo(rank + 1, score, bookId, "book");
            return ResponseResult.okResult(rankInfo);

        } catch (Exception e) {
            log.error("获取图书排名失败: bookId={}, rankingType={}", bookId, rankingType, e);
            return ResponseResult.errorResult(500, "获取图书排名失败");
        }
    }

    /**
     * 从 Redis ZSet 读取排行榜数据
     */
    private ResponseResult getRankingFromRedis(String redisKey, Integer limit, String itemType) {
        try {
            if (limit == null || limit <= 0) {
                limit = "book".equals(itemType) ? hotSize : defaultSize;
            }

            // 1. 尝试从缓存获取完整的排行榜数据（包含图书详细信息）
            String cacheKey = redisKey + ":enriched:" + limit;
            String cachedData = cacheService.get(cacheKey);
            
            if (cachedData != null && !cachedData.isEmpty()) {
                List<RankingItemVo> cachedList = JSON.parseArray(cachedData, RankingItemVo.class);
                log.info("✅ 命中缓存，直接返回排行榜: key={}, size={}", cacheKey, cachedList.size());
                return ResponseResult.okResult(cachedList);
            }
            
            log.info("❌ 缓存未命中，开始构建排行榜: key={}", cacheKey);

            // 2. 从 Redis ZSet 获取排行榜（倒序，分数高的在前）
            Set<String> topItems = cacheService.zReverseRange(redisKey, 0, limit - 1);

            log.info("从Redis获取排行榜原始数据: key={}, topItems={}", redisKey, topItems);

            if (topItems == null || topItems.isEmpty()) {
                log.warn("排行榜为空: key={}", redisKey);
                return ResponseResult.okResult(new ArrayList<>());
            }

            // 3. 构建排行榜数据并补充详细信息
            List<RankingItemVo> rankingList = new ArrayList<>();
            int rank = 1;

            log.info("开始处理{}个排行榜项目", topItems.size());
            for (String itemIdStr : topItems) {
                try {
                    // 尝试解析为整数,跳过无效数据(如 "book_002" 等测试数据)
                    Integer itemId = Integer.parseInt(itemIdStr);
                    Double score = cacheService.zScore(redisKey, itemIdStr);
                    if (score == null) {
                        score = 0.0;
                    }

                    RankingItemVo vo = RankingItemVo.builder()
                            .rank(rank++)
                            .itemId(itemId)
                            .itemType(itemType)
                            .score(score)
                            .build();

                    // 补充图书/用户的详细信息
                    if ("book".equals(itemType)) {
                        enrichBookInfo(vo, itemIdStr);
                    } else if ("user".equals(itemType)) {
                        enrichUserInfo(vo, itemIdStr);
                    }

                    rankingList.add(vo);
                } catch (NumberFormatException e) {
                    // 跳过无效的 ID 格式(如测试数据 "book_002")
                    log.warn("跳过无效的排行榜数据: itemId={}, key={}", itemIdStr, redisKey);
                }
            }

            // 4. 将补充完整信息的排行榜缓存到 Redis (5分钟过期)
            try {
                cacheService.setEx(cacheKey, JSON.toJSONString(rankingList), 300, TimeUnit.SECONDS);
                log.info("排行榜数据已缓存: key={}, size={}, ttl=300s", cacheKey, rankingList.size());
            } catch (Exception e) {
                log.warn("缓存排行榜数据失败: key={}", cacheKey, e);
            }

            log.info("获取排行榜成功: key={}, size={}", redisKey, rankingList.size());
            return ResponseResult.okResult(rankingList);

        } catch (Exception e) {
            log.error("获取排行榜失败: key={}", redisKey, e);
            return ResponseResult.errorResult(500, "获取排行榜失败");
        }
    }

    /**
     * 补充图书详细信息
     * 通过 Feign 调用 book-service 获取图书信息（仅获取必要字段：标题、作者、封面）
     */
    private void enrichBookInfo(RankingItemVo vo, String bookId) {
        try {
            Integer bookIdInt = Integer.parseInt(bookId);
            ResponseResult result = bookClient.getBookVo(bookIdInt);
            
            log.info("Feign调用book-service: bookId={}, result={}", bookId, 
                    result != null ? "code=" + result.getCode() : "null");
            
            if (result != null && result.getCode() == 200 && result.getData() != null) {
                // 解析图书信息，只设置必要字段
                JSONObject bookInfo = JSON.parseObject(JSON.toJSONString(result.getData()));
                
                log.info("解析图书信息: bookId={}, bookInfo={}", bookId, bookInfo);
                
                String title = bookInfo.getString("title");
                String author = bookInfo.getString("author");
                String cover = bookInfo.getString("cover");
                
                vo.setBookTitle(title != null ? title : "未知书名-" + bookId);
                vo.setBookAuthor(author != null ? author : "未知作者");
                vo.setBookCover(cover != null ? cover : "");
                
                log.info("图书信息设置完成: bookId={}, title={}, author={}, cover={}", 
                        bookId, vo.getBookTitle(), vo.getBookAuthor(), 
                        vo.getBookCover() != null && vo.getBookCover().length() > 50 ? 
                        vo.getBookCover().substring(0, 50) + "..." : vo.getBookCover());
            } else {
                // 如果获取失败,使用默认值
                vo.setBookTitle("图书-" + bookId);
                vo.setBookAuthor("未知作者");
                vo.setBookCover("");
                log.warn("Feign 调用 book-service 失败或返回空: bookId={}, result={}", bookId, result);
            }
        } catch (NumberFormatException e) {
            log.error("图书ID格式错误: bookId={}", bookId, e);
            vo.setBookTitle("图书-" + bookId);
            vo.setBookAuthor("未知作者");
            vo.setBookCover("");
        } catch (Exception e) {
            log.error("获取图书详细信息失败: bookId={}", bookId, e);
            vo.setBookTitle("图书-" + bookId);
            vo.setBookAuthor("未知作者");
            vo.setBookCover("");
        }
    }
    
    /**
     * 补充用户详细信息
     * 注意: 目前用户排行榜功能暂未使用,此方法为预留接口
     */
    private void enrichUserInfo(RankingItemVo vo, String userId) {
        try {
            // TODO: 用户信息需要从 user-service 获取或实现用户信息缓存
            vo.setUsername("用户-" + userId);
            log.debug("用户排行榜暂未实现详细信息获取: userId={}", userId);
        } catch (Exception e) {
            log.error("获取用户详细信息失败: userId={}", userId, e);
            vo.setUsername("用户-" + userId);
        }
    }

    /**
     * 根据排行榜类型获取 Redis Key
     */
    private String getRankingKey(String rankingType) {
        if (rankingType == null) {
            return null;
        }

        switch (rankingType) {
            case "book_hot":
                return KafkaConfig.Keys.BOOK_HOT_RANKING;
            case "book_read":
                return KafkaConfig.Keys.BOOK_READ_RANKING;
            case "book_collect":
                return KafkaConfig.Keys.BOOK_COLLECT_RANKING;
            case "book_like":
                return KafkaConfig.Keys.BOOK_LIKE_RANKING;
            case "user_reading_time":
                return KafkaConfig.Keys.USER_READING_TIME_RANKING;
            case "user_active":
                return KafkaConfig.Keys.USER_ACTIVE_RANKING;
            default:
                return null;
        }
    }

    /**
     * 构建排名信息
     */
    private Map<String, Object> buildRankInfo(Long rank, Double score, Integer itemId, String itemType) {
        Map<String, Object> rankInfo = new HashMap<>();
        rankInfo.put("itemId", itemId);
        rankInfo.put("itemType", itemType);
        rankInfo.put("rank", rank);
        rankInfo.put("score", score);
        rankInfo.put("inRanking", rank != null);
        
        return rankInfo;
    }

    // ========== 笔记排行榜 ==========

    @Override
    public ResponseResult getRealtimeHotNotes(Integer limit) {
        return getRankingFromRedis(KafkaConfig.Keys.NOTE_REALTIME_HOT_RANKING, limit, "note");
    }

    @Override
    public ResponseResult getHotNotes(Integer limit) {
        return getRankingFromRedis(KafkaConfig.Keys.NOTE_HOT_RANKING, limit, "note");
    }

    // ========== 书评排行榜 ==========

    @Override
    public ResponseResult getRealtimeHotReviews(Integer limit) {
        return getRankingFromRedis(KafkaConfig.Keys.REVIEW_REALTIME_HOT_RANKING, limit, "review");
    }

    @Override
    public ResponseResult getHotReviews(Integer limit) {
        return getRankingFromRedis(KafkaConfig.Keys.REVIEW_HOT_RANKING, limit, "review");
    }

    @Override
    public ResponseResult getBookHotNotes(Integer bookId, Integer limit) {
        // TODO: 从笔记服务获取指定图书的笔记列表，然后按Redis中的热度排序
        // 这里需要调用 Feign 与 note-service 交互
        log.info("获取图书热门笔记: bookId={}, limit={}", bookId, limit);
        return ResponseResult.okResult("功能开发中");
    }

    @Override
    public ResponseResult getBookHotReviews(Integer bookId, Integer limit) {
        // TODO: 从书评服务获取指定图书的书评列表，然后按Redis中的热度排序
        // 这里需要调用 Feign 与 review-service 交互
        log.info("获取图书热门书评: bookId={}, limit={}", bookId, limit);
        return ResponseResult.okResult("功能开发中");
    }
}

